package com.cn.kafka;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.List;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import com.cn.component.SnowFlake;
import com.cn.domain.OrderEntity;
import com.cn.service.OrderService;

/**
 * 实际消费者,继承了ShutdownableThread ,要多加几个消费者直接继承实现即可
 *
 * @create 2017-11-06 12:42
 * @update 2017-11-06 12:42
 **/
public class Consumer extends  Thread {

    @Override
	public void destroy() {
		// TODO Auto-generated method stub
    	this.stop=true;
		super.destroy();
	}

	@Override
	public void interrupt() {
		// TODO Auto-generated method stub
    	this.stop=true;
		super.interrupt();
	}

	/**
     * kafka 消费者
     */
    private  KafkaConsumer consumer;

    /**
     *  topic
     */
    private  String topic;

    /**
     *  组id
     */
    private  String groupId;
    
    private boolean stop;
    private OrderService orderService;

    public Consumer(ConsumerGroup consumerGroup,OrderService orderService) {
      
        this.consumer =  consumerGroup.getConsumer();
        this.topic = consumerGroup.getTopic();
        this.groupId = consumerGroup.getA_groupId();
        this.stop=false;
        this.orderService=orderService;
    }

    /**
     *  * 监听主题,有消息就读取
     * 从kafka里面得到数据后,具体怎么去处理. 如果需要开启kafka处理消息的广播模式,多个监听要监听不同的group,
     * 即方法上的注解@KafkaListener里的group一定要不一样.如果多个监听里的group写的一样,就会造成只有一个监听能处理其中的消息,
     * 另外监听就不能处理消息了.也即是kafka的分布式消息处理方式.
     * 在同一个group里的监听,共同处理接收到的消息,会根据一定的算法来处理.如果不在一个组,但是监听的是同一个topic的话,就会形成广播模式
     */
    @Override
    public void run() {
        consumer.subscribe(Collections.singletonList(this.topic));
        List<OrderEntity> orderList = new ArrayList<OrderEntity>();
       	SnowFlake snowFlake=new SnowFlake(0,0);
        while(!stop){
	        ConsumerRecords<Integer, String> records = consumer.poll(1000);
	        for (ConsumerRecord<Integer, String> record : records) {
	            System.out.println("Thread: "+Thread.currentThread().getName()
	                    +"Received message: (" + this.groupId + ", " + record.value() + ") at offset "
	                    + record.offset()+" partition : "+records.partitions());
	            String[] info = record.value().split("_");
	            OrderEntity order = new OrderEntity();
	   		      order.setModifyTime(new Date());
	   		      order.setProductId(Long.parseLong(info[0]));
	   		      order.setNum(Integer.parseInt(info[1]));
	   		      order.setUserId(Long.parseLong(info[2]));
	   		      order.setOrderNo(snowFlake.nextId()+"");
	   	      	  orderList.add(order);
	        }
	        
	        if(orderList.size() > 1200 || (orderList.size()>0 && records.isEmpty())){
	        	this.orderService.insertBatchOrder(orderList);
	        	orderList.clear();
	        }
        }
    }
}